package com.ksyun.campus.metaserver.services;

import com.ksyun.campus.common.constant.ZkConstant;
import com.ksyun.campus.common.constant.api.DataServerApi;
import com.ksyun.campus.common.domain.*;
import com.ksyun.campus.common.util.ByteConvertUtil;
import com.ksyun.campus.metaserver.util.FSHttpUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Service
@Slf4j
public class FsckServices {

    @Autowired
    private CuratorFramework curatorFramework;

    @Autowired
    private MetaService metaService;

    private HashMap<String, StatInfo> map;

    //@Scheduled(cron = "0 0 0 * * ?") // 每天 0 点执行
    @Scheduled(fixedRate = 50000) // 每隔 30 秒执行一次
    public void fsckTask() {
        map = new HashMap<>(100);
        try {
            scan(ZkConstant.META_DATA_PATH);
        } catch (Exception e) {
            e.printStackTrace();
        }
        for (Map.Entry<String, StatInfo> entry : map.entrySet()){
            StatInfo statInfo = entry.getValue();
            List<ReplicaData> replicaData = statInfo.getReplicaData();
            if(replicaData.size() != 0 && replicaData.size() < 3){
                //获取新的节点
                List<String> newNode = getNewNode(replicaData, 3 - replicaData.size());
                //选一个正常可用分数据节点请求修复
                repair(newNode, replicaData.get(0).dsNode, entry.getKey());
                //更新元数据
                updateMeta(entry, statInfo, replicaData, newNode);
            }
        }
    }

    private void updateMeta(Map.Entry<String, StatInfo> entry, StatInfo statInfo, List<ReplicaData> replicaData, List<String> newNode) {
        for(String  node : newNode){
            ReplicaData addRep = new ReplicaData();
            DataServer dataServerInZk = getDataServerInZk(ZkConstant.DATA_SERVER + "/" + node);
            addRep.dsNode = node;
            addRep.id = dataServerInZk.getId();
            addRep.path = statInfo.getPath();
            replicaData.add(addRep);
        }
        statInfo.setReplicaData(replicaData);
        try {
            if (curatorFramework.checkExists().forPath(entry.getKey()) != null) {
                curatorFramework.setData().forPath(entry.getKey(), ByteConvertUtil.toByte(statInfo));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 修复
     * @param newNode
     * @param queryUrl 正常可用的数据节点
     * @param cpath 完全路径
     */
    private void repair(List<String> newNode, String queryUrl, String cpath ){
        RepairBody repairBody = new RepairBody();
        repairBody.setRepairNode(newNode);
        repairBody.setCpath(cpath);
        FSHttpUtils.post(queryUrl + DataServerApi.REPAIR_DATA, null, "", repairBody);
    }

    /**
     * 排除现有正常的节点 选取新的正常的节点
     * @param exclude 排除的节点
     * @param  num  需要节点的数量
     * @return 新的 address
     */
    private List<String> getNewNode(List<ReplicaData> exclude, int num){
        List<String> res =  new ArrayList<>();
        List<String> excludeStr= new ArrayList<>();
        for(ReplicaData replicaData : exclude){
            excludeStr.add(replicaData.dsNode);
        }
        List<DataServer> dataList = metaService.getDataList();
        for (DataServer dataServer : dataList){
            String address = dataServer.getIp() + ":" + dataServer.getPort();
            if(!excludeStr.contains(address)){
                res.add(address);
            }
        }
        List<String> dataServers = res.stream().limit(num).collect(Collectors.toList());
        return dataServers;
    }


    private void scan(String parent) throws Exception {
        List<String> list = curatorFramework.getChildren().forPath(parent);
        for(String path : list){
            String cPath = parent + "/" + path;
            StatInfo statInfo = getStatInfo(cPath);
            if(statInfo.type.equals(FileType.Directory) || statInfo.type.equals(FileType.Volume) ){
                scan(cPath);
            }else {
                List<ReplicaData> replicaData = statInfo.getReplicaData();
                checkReplica(replicaData, cPath);
                statInfo.setReplicaData(replicaData);
                map.put(cPath,statInfo);
            }
        }
    }

    /**
     * 检查副本信息
     * @param replicaData
     * @param cpath
     */
    private void checkReplica(List<ReplicaData> replicaData, String cpath) {
        ListIterator<ReplicaData> iterator = replicaData.listIterator();
        while (iterator.hasNext()){
            ReplicaData next = iterator.next();
            String url = next.dsNode + DataServerApi.CHECK_REPLICA;
            if(!FSHttpUtils.checkReplica(url, cpath)){
                iterator.remove();
            }
        }
    }


    private StatInfo getStatInfo(String metapath) throws Exception {
        byte[] bytes = curatorFramework.getData().forPath(metapath);
        return (StatInfo) ByteConvertUtil.toObj(bytes, StatInfo.class);
    }


    private DataServer getDataServerInZk(String nodePath) {
        byte[] bytes = new byte[0];
        try {
            bytes = curatorFramework.getData().forPath(nodePath);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return (DataServer) ByteConvertUtil.toObj(bytes, DataServer.class);
    }




}
